package com.ehotting.mqbd.core.mq.consumer.action;

import com.ehotting.eaf.core.snow.SnowflakeUtil;
import com.ehotting.mqbd.api.bean.MQDataAble;
import com.ehotting.mqbd.api.bean.MqApplicationContext;
import com.ehotting.mqbd.api.enums.DirectionEnum;
import com.ehotting.mqbd.api.enums.TopicEnum;
import com.ehotting.mqbd.core.logger.LOGGER_KEY;
import com.ehotting.mqbd.core.logger.MqMsgLogger;
import com.ehotting.mqbd.core.mq.listener.MQListenerConcurrently;
import com.ehotting.mqbd.core.mq.listener.MQListenerOrderly;
import com.ehotting.mqbd.core.manager.ModuleManager;
import com.ehotting.mqbd.core.sync.MqSyncFutureMapHelper;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.MDC;

public class ConsumerResponseAction extends ConsumerActionAbstract {

    private MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper;
    private MqApplicationContext mqApplicationContext;
    private TopicEnum topic;
    private DirectionEnum direction;
    private String solverIp;

    public ConsumerResponseAction(MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper,
                                  MqApplicationContext mqApplicationContext,
                                  TopicEnum topic,
                                  DirectionEnum direction,
                                  ModuleManager moduleManager, MqMsgLogger mqMsgLogger){
        this.mqSyncFutureMapHelper = mqSyncFutureMapHelper;
        this.mqApplicationContext = mqApplicationContext;
        this.topic = topic;
        this.direction = direction;
        this.moduleManager = moduleManager;
        this.mqMsgLogger = mqMsgLogger;
    }

    public ConsumerResponseAction(MqSyncFutureMapHelper<MQDataAble> mqSyncFutureMapHelper,
                                  MqApplicationContext mqApplicationContext,
                                  TopicEnum topic,
                                  DirectionEnum direction,
                                  String solverIp,
                                  ModuleManager moduleManager, MqMsgLogger mqMsgLogger){
        this.mqSyncFutureMapHelper = mqSyncFutureMapHelper;
        this.mqApplicationContext = mqApplicationContext;
        this.topic = topic;
        this.direction = direction;
        this.solverIp = solverIp;
        this.moduleManager = moduleManager;
        this.mqMsgLogger = mqMsgLogger;
    }

    @Override
    public String getModule() {
        return mqApplicationContext.getModule();
    }

    @Override
    public void invoke(MessageExt msg) {
        logger.info("Start invoke.");
        MDC.put(LOGGER_KEY.TRACE_ID, SnowflakeUtil.nextId64());
        MDC.put(LOGGER_KEY.MODULE,mqApplicationContext.getModule());
        String str =  new String(msg.getBody());
        MQDataAble data = convert(str,msg.getMsgId());
        mqSyncFutureMapHelper.putAndNotify(data.getSerial(), data);
        extInvoke(data);
        logger.info("Finish invoke.");
    }

    public String getHost(){
        return mqApplicationContext.getLocalHost();
    }

    @Override
    public DirectionEnum getDirection() {
        return direction;
    }

    @Override
    public TopicEnum getTopic() {
        return topic;
    }

    @Override
    public String getSolverIp() {
        return solverIp;
    }

    @Override
    public void extInvoke(MQDataAble data) {

    }

    @Override
    public MessageListener getMessageListener() {
        return this.getTopic()== TopicEnum.BLOCK?
                new MQListenerOrderly(this):
                new MQListenerConcurrently(this);
    }

    @Override
    public MqApplicationContext getMqApplicationContext() {
        return mqApplicationContext;
    }

    public void setMqMsgLogger(MqMsgLogger mqMsgLogger) {
        this.mqMsgLogger = mqMsgLogger;
    }
}
